package cn.doitedu.service;

import cn.doitedu.dao.*;
import cn.doitedu.pojo.*;
import cn.doitedu.utils.MatchUtils;
import cn.doitedu.utils.TimeBoundUtils;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.List;
import java.util.Map;

public class MatchService {

    private UserProfileDao userProfileDao;
    private HistoryEventDao historyEventDao;
    private EventStateDaoImpl eventStateDao;

    //初始化连接
    public void init(ParameterTool parameterTool, ListState<EventBean> listState, MapState<String, String> cacheState) throws Exception {
        //目前DAO的实现是先new，以后可以从配置文件中读取实现类的名称，在通过反射创建
        //userProfileDao = new UserProfileHbaseDaoImpl();
        String profileDaoImpl = parameterTool.getRequired("profile.dao.impl");
        userProfileDao = (UserProfileDao) Class.forName(profileDaoImpl).newInstance();
        userProfileDao.init(parameterTool);
        //实现类的new 不要写死在程序中，而是通过反射创建
        String eventDaoImpl = parameterTool.getRequired("event.dao.impl");
        historyEventDao = (HistoryEventDao) Class.forName(eventDaoImpl).newInstance();
        //historyEventDao = new HistoryEventClickHouseDaoImpl();
        historyEventDao = new HistoryEventDorisDaoImpl();
        historyEventDao.init(parameterTool, cacheState);

        eventStateDao = new EventStateDaoImpl();
        eventStateDao.init(parameterTool, listState);

    }

    //关闭连接
    public void close() throws Exception {
        userProfileDao.close();
        historyEventDao.close();
    }

    //进行匹配的方法
    public boolean isMath(LogBean bean, RuleConditionV2 ruleCondition) throws Exception {

        boolean matchEvent = isMatchEvent(bean, ruleCondition.getTriggerEvent());
        if (!matchEvent) {
            return false;
        }
        //并且的关系
        boolean matchProfile = isMatchProfile(bean, ruleCondition.getUserProfiles());
        if (!matchProfile) {
            return false;
        }
        //并且的关系
        boolean matchEventCount = isMatchEventCount(bean, ruleCondition.getCombineConditions());

        return matchEventCount;
    }

    //匹配触发事件和行为参数
    public boolean isMatchEvent(LogBean bean, EventParam eventParam) {
        //完成匹配触发事件的逻辑
        //eventParam中封装着事先定义好的触发事件ID和属性
        String eventId = eventParam.getEventId();
        Map<String, String> props = eventParam.getProps();
        //用户的实际的事件ID
        //用户实际的行为属性
        Map<String, String> properties = bean.getProperties();
        if (eventId.equals(bean.getEventId())) {
            //匹配行为属性
            for (Map.Entry<String, String> entry : props.entrySet()) {
                //事先定义好的行为属性名称和值
                String targetKey = entry.getKey();
                String targetVal = entry.getValue();
                //根据事先定义好的属性key，取出实际key的value
                String val = properties.get(targetKey);
                if (!targetVal.equals(val)) {
                    return false;
                }
            }
        } else {
            return false;
        }
        return true;
    }

    //匹配用户画像
    public boolean isMatchProfile(LogBean bean, Map<String, String> userProfiles) throws Exception {

        //调用Hbase的DAO进行查询
        Result result = userProfileDao.queryUserProfile(bean, userProfiles);
        //在Service层做逻辑判断
        if (result != null) {
            for (String tag : userProfiles.keySet()) {
                String val = new String(result.getValue(Bytes.toBytes("f"), Bytes.toBytes(tag)));
                //取出事先定义好的该tag的值
                String targetVal = userProfiles.get(tag);
                if (!targetVal.equals(val)) {
                    return false;
                }
            }
        } else {
            return false;
        }
        return true;
    }

    //匹配行为次数和行为序列
    public boolean isMatchEventCount(LogBean bean, List<CombineCondition> combineConditions) throws Exception {

        //调用ClickHouse或Doris的DAO进行数据查询
        //historyEventDao....
        for (CombineCondition combineCondition : combineConditions) {
            //时间边界
            //为了以后缓存的命中率更高，所以时间边界最好一小时变一次（不要频繁变化，频繁变化，查询条件就变化，就没法复用缓存了）
            //long timeBound = bean.getTimeStamp() - 1000 * 60 * 60 * 2;
            Long timeBound = TimeBoundUtils.getTimeBound(bean.getTimeStamp(), 2);
            Long startTime = combineCondition.getStartTime();
            Long endTime = combineCondition.getEndTime();
            if (startTime >= timeBound) {
                //（近期数据）从状态中查询
                String eventStr = eventStateDao.queryEventStr(bean, combineCondition);
                //使用正则进行匹配
                int count = MatchUtils.matchCount(combineCondition.getRegex(), eventStr);
                if(count < combineCondition.getCount()) {
                    return false;
                }
            } else if (endTime <= timeBound) {
                //（远期数据）从doris中查询
                String eventStr = historyEventDao.queryEventStr(bean, combineCondition);
                int count = MatchUtils.matchCount(combineCondition.getRegex(), eventStr);
                if(count < combineCondition.getCount()) {
                    return false;
                }
            } else {
                //跨边界查询
                combineCondition.setStartTime(timeBound);
                String eventStr = eventStateDao.queryEventStr(bean, combineCondition);
                int count = MatchUtils.matchCount(combineCondition.getRegex(), eventStr);
                if(count < combineCondition.getCount()) {
                    //再查询Doris
                    combineCondition.setStartTime(startTime);
                    combineCondition.setEndTime(timeBound);
                    String eventStr1 = historyEventDao.queryEventStr(bean, combineCondition);
                    count = MatchUtils.matchCount(combineCondition.getRegex(), eventStr1 + eventStr);
                    if(count < combineCondition.getCount()) {
                        return false;
                    }
                }
            }
        }
        return true;
    }

}
